package com.rose.datax.dataxadmin.manager.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.rose.datax.dataxadmin.entity.MigrationTable;
import com.rose.datax.dataxadmin.entity.MigrationTask;
import com.rose.datax.dataxadmin.mapper.MigrationTableMapper;
import com.rose.datax.dataxadmin.mapper.MigrationTaskMapper;
import com.rose.datax.dataxadmin.utils.ArrayUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;

@Service
@Slf4j
public class TransferTask {

    @Resource
    private MigrationTableMapper tableMapper;

    @Value("${datax.home}")
    private String dataxHome;

    @Resource
    private MigrationTaskMapper taskMapper;

    private List<MigrationTable> tables;

    public void transfer(long taskId){
        long start = System.currentTimeMillis();
        tables = tableMapper.selectList(new QueryWrapper<MigrationTable>().eq("task_id", taskId));

        //把数据分成8个数组执行
        List<List<MigrationTable>> splitTables = ArrayUtils.splitArray(tables);
        MigrationTask task = taskMapper.selectById(taskId);
        task.setStatus("运行中");
        taskMapper.updateById(task);

        CountDownLatch countDownLatch = new CountDownLatch(splitTables.size());
        //CPU 核数
        int cpuNum = Runtime.getRuntime().availableProcessors();
        ExecutorService executor = new ThreadPoolExecutor(cpuNum, cpuNum * 2, 60000L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

        for(List<MigrationTable> list: splitTables){
            executor.submit(new DataXThread(list, task, countDownLatch));
        }
        //等待所有线程执行完毕
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        executor.shutdown();
        log.info("task exec over!");
        long time = System.currentTimeMillis() - start;
        task.setRunTime(time/1000);
        task.setStatus("完成");
        task.setGmtModify(new Date());
        taskMapper.updateById(task);
    }

    private boolean executeCommand(String command, MigrationTable table) throws IOException{
        boolean hasException = false;
        Process process = Runtime.getRuntime().exec(command);
        BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
        String line = null;
        while ((line = reader.readLine()) != null) {
            if (!hasException && (line.contains("DataXException") || line.contains("SQLException"))){
                hasException = true;
            }
            readDataxOutput(line, table);
        }
        return hasException;
    }

    /**
     * 分析datax 任务执行情况
     * @param line
     * @param table
     */
    private void readDataxOutput(String line, MigrationTable table){
        if (line!=null && line.contains("  :  ")){
            String[] resultTemp = line.split("  :  ");
            String value = resultTemp[1].trim();
            if (value.matches("\\d+s")) {
                log.info("任务耗时: {}", value);
                //table.setCostTime(value);
            } else if (value.matches("\\d+[A-Z]*B\\/s")){
                log.info("任务平均流量: {}", value);
            } else if (value.matches("\\d+rec\\/s")){
                log.info("记录写入速度: {}", value);
            }
        }else if (line!=null && line.contains("used")){
            try {
                int count = Integer.parseInt(line.substring(line.indexOf("used") + 5, line.indexOf("ms") - 1).trim());
                table.setCostTime(count + "ms");
            }catch (Exception e){
            }
        } else if (line!=null && line.contains("StandAloneJobContainerCommunicator")){
            int count = Integer.parseInt(line.substring(line.indexOf("Total")+5, line.indexOf("records")).trim());
            log.info("读出记录总数: {}", count);
            table.setReadRecords(count);
            int countFail = Integer.parseInt(line.substring(line.indexOf("Error")+5, line.indexOf("records", line.indexOf("Error"))).trim());
            log.info("读写失败总数: {}", countFail);
            table.setWriteFail(countFail);
            String avgFlow = line.substring(line.indexOf("Speed")+5, line.indexOf("B/s")+3).trim();
            log.info("任务平均流量: {}", avgFlow);
            String speed = line.substring(line.indexOf(",", line.indexOf("B/s"))+1, line.indexOf("records/s")+9).trim();
            log.info("记录写入速度: {}", speed);
        }
        tableMapper.updateById(table);
    }

    private String getCommand(long taskId, String tableName){
        String jobFolder = dataxHome;
        if(jobFolder.endsWith("/")){
            jobFolder = jobFolder +"job/task-"+taskId+"/";
        }else{
            jobFolder = jobFolder +"/job/task-"+taskId+"/";
        }
        return String.format("python %s/bin/datax.py %s%s.json", dataxHome, jobFolder, tableName);
    }


    class DataXThread extends Thread{

        private List<MigrationTable> tables;
        private CountDownLatch countDownLatch;
        private MigrationTask task;

        private DataXThread(List<MigrationTable> tables, MigrationTask task, CountDownLatch countDownLatch){
            this.tables = tables;
            this.task = task;
            this.countDownLatch = countDownLatch;
        }

        public void run() {
            for(MigrationTable table: tables){
                if(table.getRows() == 0){
                    log.info("table {} empty", table.getTableName());
                    table.setStatus("完成");
                    tableMapper.updateById(table);
                    continue;
                }
                table.setStatus("运行中");
                tableMapper.updateById(table);
                long start = System.currentTimeMillis();
                String command = getCommand(task.getId(), table.getTableName());
                log.info("command: -> {}", command);
                //执行迁移任务
                boolean hasException = false;
                try {
                    hasException = executeCommand(command, table);
                    if(task.isErrorBreak() && hasException){
                        break;
                    }
                } catch (IOException e) {
                    log.error(e.getMessage(), e);
                }
                long time = System.currentTimeMillis() - start;
                table.setStatus("完成");
                table.setCostTime(time/1000+"s");
                tableMapper.updateById(table);
            }
            countDownLatch.countDown();
        }
    }

}
